package com.dxy.chatserver.service;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.json.JSON;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dxy.chatserver.db.entity.AcceptLineMsgResponse;
import com.dxy.chatserver.db.entity.AcceptLineMsgResponseBodies;
import com.dxy.chatserver.db.entity.AcceptLineMsgResponseUser;
import com.dxy.chatserver.mapper.ChatMsgMapper;
import com.mongodb.client.MongoCursor;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;

@Service
@Slf4j
public class MoveService {

    @Autowired
    private ChatMsgMapper chatMsgMapper;
    @Autowired
    private EsService esService;

    //按照一个月的去处理
    /**
     * 2022-06-13 00:00:00 1655049600000
     * 2022-05-01 00:00:00 1651334400000
     */

    public void saveData(String startTime,String endTime) {
        try {
            Set<String> collect = chatMsgMapper.getDbNameList(null);
            if (collect.isEmpty()) {
                return;
            }
            log.info("转移数据:{}库条数", collect.size());
            collect.forEach(o -> {
                    doSave(o,startTime,endTime);
            });
            log.info("所有库处理完成");
        }catch (Exception e){
            log.info("出现异常 提前结束 检查问题 : {}",e);
            return;
        }
    }

    private void doSave(String o,String startTime,String endTime){
        log.info("当前线程 {}   开始处理表:{}",Thread.currentThread().getName(),o);
        if(!StringUtils.hasText(startTime) || !StringUtils.hasText(endTime)){
            return;
        }
        Long start = Long.parseLong(startTime);
        Long end = Long.parseLong(endTime);
        chatMsgMapper.getCount(o,start,end);
        MongoCursor<Document> dbObject = chatMsgMapper.findDBObject(o, start, end);
        AtomicInteger count1 = new AtomicInteger(0);
        List<AcceptLineMsgResponse> acceptLineMsgResponsesSave = new ArrayList<>();
        dbObject.forEachRemaining(document->{
            AcceptLineMsgResponse acceptLineMsgResponse = JSONUtil.toBean(document.toJson(), AcceptLineMsgResponse.class);
            String s = document.toJson();
            if(!ObjectUtil.isEmpty(s)){
                JSONObject json = JSONUtil.parseObj(s);
                if(json!=null){

                    Object fromUser = json.get("fromUser");
                    if(fromUser!=null){
                        JSONObject jsonObject = JSONUtil.parseObj(fromUser);
                        if(!jsonObject.isEmpty()){
                            Object id = jsonObject.get("_id");
                            if(!ObjectUtil.isEmpty(id)){
                                acceptLineMsgResponse.getFromUser().setId(Long.parseLong(id.toString()));
                            }
                        }
                    }

                    Object toUser = json.get("toUser");
                    if(toUser!=null){
                        JSONObject jsonObject = JSONUtil.parseObj(toUser);
                        if(!jsonObject.isEmpty()){
                            Object id = jsonObject.get("_id");
                            if(!ObjectUtil.isEmpty(id)){
                                acceptLineMsgResponse.getToUser().setId(Long.parseLong(id.toString()));
                            }
                        }
                    }
                }
            }
            acceptLineMsgResponsesSave.add(acceptLineMsgResponse);
            int i = count1.incrementAndGet();
            if(i==10000){
                //提交一次
                esService.addBatch(acceptLineMsgResponsesSave,o);
                count1.set(0);
                log.info("满足次提交:{}",acceptLineMsgResponsesSave.size());
                acceptLineMsgResponsesSave.clear();
            }
        });
        if(!acceptLineMsgResponsesSave.isEmpty()){
            log.info("最后一次提交:{}",acceptLineMsgResponsesSave.size());
            esService.addBatch(acceptLineMsgResponsesSave,o);
        }
        log.info("线程:{}执行完成",Thread.currentThread().getName());
//                List<AcceptLineMsgResponse> acceptLineMsgResponses = chatMsgMapper.find(o,1654012800000L,1654617600000L);
//                log.info("查询条数{}", acceptLineMsgResponses.size());
//                List<List<AcceptLineMsgResponse>> splitList = ListUtil.split(acceptLineMsgResponses, 200);
//                log.info("需要拆分为:{}组",splitList.size());
//                splitList.forEach(list->{
//                    //异步处理
////                    TsService.batchWriteRow(list);
//                    //es落库处理
//                    esService.addBatch(list);
//                });
    }
}
